package com.gitee.cnlongs.cnlongmq.server.threads.engine;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSON;
import com.gitee.cnlongdb.basic.entity.datasync.CnLongDataSyncRequest;
import com.gitee.cnlongdb.basic.entity.datasync.CnLongDataSyncResponse;
import com.gitee.cnlongdb.basic.enums.sysnc.LongdbDataSyncEnms;
import com.gitee.cnlongdb.basic.utils.RandomUtil;
import com.gitee.cnlongdb.server.connect.LongDbConnect;
import com.gitee.cnlongdb.server.datasync.DataSyncConnectMessage;
import com.gitee.cnlongs.cnlongmq.basic.constant.CnLongMqConstant;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongMessageQueue;
import com.gitee.cnlongs.cnlongmq.basic.entitiy.request.CnLongRequest;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqDataSync;
import com.gitee.cnlongs.cnlongmq.basic.enums.CnLongMqStatus;
import com.gitee.cnlongs.cnlongmq.server.LongMqServer;
import com.gitee.cnlongs.cnlongmq.server.threads.SequeueHeartBeatHandler;

/**
 * 系统名称：SequenceAvailableMonitorThread.java
 * 模块名称：
 * 模块描述：队列可用性监听
 * 功能列表：
 * 模块作者：ZOUYONG
 * 开发时间：2019年7月27日 上午10:35:50
 * 模块路径：com.gitee.cnlongs.cnlongmq.server.threads.engine.SequenceAvailableMonitorThread
 * 更新记录：
 */
public class SequenceAvailableMonitorThread extends Thread {

	private static final Logger logger = (Logger) LoggerFactory.getLogger(SequenceAvailableMonitorThread.class);
	
	private LongDbConnect connect;
	
	public SequenceAvailableMonitorThread(LongDbConnect connect) {
		this.connect = connect;
	}
	
	@Override
	public void run() {
		Date currDate = new Date();
		
		// 对同步队列信息的缓存数据进行清理
		try {
			Map<String, Long> alreadyOpertorCustomerMap = 
					LongMqServer.getExecuter().getLongMqConfig().getAlreadyOpertorCustomerMap();
			for(Iterator<String> iter = alreadyOpertorCustomerMap.keySet().iterator(); iter.hasNext(); ) {
				String operatorCode = iter.next();
				Long operatorTime = alreadyOpertorCustomerMap.get(operatorCode);
				// 队列同步操作的缓存数据，超出15秒，则进行清理
				if(null != operatorTime && currDate.getTime() - 
						operatorTime >= 3 * CnLongMqConstant.heartbeatCheckIntervalTime) {
					alreadyOpertorCustomerMap.remove(operatorCode);
				}
			}
		} catch(Exception e) {
			logger.error("清理队列同步操作的缓存数据异常：{}", e.getMessage());
		}
		
		// 对队列进行心跳检查
		try {
			// 获取需要尝试心跳连接的队列
			List<CnLongMessageQueue> sequeueList = 
					LongMqServer.getExecuter().getLongMqConfig().getAppRelationSequeueList();
			for (CnLongMessageQueue queue : sequeueList) {
				// 是否到达心跳检测时间
				if(null != queue.getHeartBeatData() && currDate.getTime() - 
						queue.getHeartBeatData().getTime() < CnLongMqConstant.heartbeatCheckIntervalTime) {
					continue;
				}
				queue.setHeartbeatConnectNumber(queue.getHeartbeatConnectNumber()+1);
				queue.setHeartBeatData(new Date());
				// 定义基础请求对象
				CnLongRequest request = new CnLongRequest();
				request.setIp(LongMqServer.getExecuter().getLongMqConfig().getLongMqIp());
				request.setPort(LongMqServer.getExecuter().getLongMqConfig().getLongMqPort());
				// 定义消息通信请求对象
				CnLongDataSyncRequest dataSync = new CnLongDataSyncRequest();
				dataSync.setMessageType(CnLongMqDataSync.checkSeqHeartBeat.name());
				dataSync.setClassPath(SequeueHeartBeatHandler.class.getName());
				dataSync.setMethodName("receiveHeartBeatRequest");
				dataSync.setParamClas(new String[]{ request.getClass().getName() });
				dataSync.setParamObject(new String[]{ JSON.toJSONString(request) });
				dataSync.setOperThreadId(RandomUtil.getCharAndNumr(20));
				// 开始发送netty消息，通知关联队列方进行队列更换
				CnLongDataSyncResponse response = DataSyncConnectMessage
						.noticeDataSyncSend(queue.getLongMqIp(), Integer.valueOf(queue.getLongMqPort()), dataSync);
				if(LongdbDataSyncEnms.success.name().equals(response.getReturnCode())) {
					queue.setLongMqStatus(CnLongMqStatus.active.name());
					queue.setHeartbeatConnectNumber(0);
				}
				else if(queue.getHeartbeatConnectNumber() >= 3) {
					queue.setLongMqStatus(CnLongMqStatus.stop.name());
					queue.setStopDate(new Date());
					queue.setHeartbeatConnectNumber(0);
				}
			}
		}
		catch(Exception e) {
			logger.error(e.getMessage());
		}
	}
	
}
